/* Copyright (C) 2008  Romolo Camplani, Marco Fortunato, Marco Marelli, Guido Rota, Fabio A. Schreiber et al.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 */
package org.dei.perla.utils.pipe;

import java.util.LinkedList;
import java.util.Queue;

import org.dei.perla.component.Component;
import org.dei.perla.utils.waiter.Waitable;

/**
 * Coda generica di messaggi. La coda gestisce la sincronizzazione tra i produttori 
 * (anche più di uno) ed il consumatore (sempre e solo uno).
 * Le operazioni di estrazione dalla coda sono non bloccanti (viene ritornato null se 
 * non ci sono dati in coda) mentre le operazioni di inserimento sono bloccanti (nel
 * caso in cui la coda risulti piena).
 */
public class Pipe<E> extends Component {

	/**
	 * Lista dei messaggi presenti nella Pipe
	 */
	private final Queue<E> pMessageQueue;
	
	/**
	 * Numero massimo di messaggi che possono essere accodati (0 = infinito)
	 */
	private final int pMessageQueueMaxLenght;
	
	/**
	 * Oggetto Waitable per notificare ad un eventuale Waiter la presenza di 
	 * messaggi nella coda
	 */
	private final Waitable<Pipe<E>> pWaitable;

	/**
	 * Costruttore (costruisce una pipe illimitata)
	 */
	public Pipe(String parName) {
		this(parName, 0);
	}
	
	/**
	 * Costruttore
	 * @param parMessageQueueLenght Massimo numero di messaggi che la Pipe potrà contenere (0 = infinito)
	 */
	public Pipe(String parName, int parMessageQueueMaxLenght) {	
		// Imposta il nome del componente
		super(parName);
		
		// Memorizza il massimo numero di messaggi che la Pipe potrà contenere
		if(parMessageQueueMaxLenght < 0) {
			parMessageQueueMaxLenght = 0;
		}
		pMessageQueueMaxLenght = parMessageQueueMaxLenght;
		
		// Istanzia la coda in cui verranno inseriti i messaggi
		pMessageQueue = new LinkedList<E>();
		
		// Istanzia l'oggetto Waitable
		pWaitable = new Waitable<Pipe<E>>(this);		
	}
	
	/**
	 * Avvia il componente pipe
	 */
	@Override
	public void start() {
		synchronized (this.pMessageQueue) {
			if(!isStarted()) {
				// Segnala che la pipe è avviata
				setStarted(true);
			}
		}
	}
	
	/**
	 * Arresta il componente pipe
	 */
	@Override
	public void stop() {
		synchronized (pMessageQueue) {
			if(isStarted()) {
				// Svuota la pipe
				pMessageQueue.clear();
				
				// Risveglia eventuali thread in attesa
				pMessageQueue.notifyAll();
				
				// Resetta il segnale nell'oggetto Waitable
				pWaitable.resetWaiter();
				
				// Segnala che la pipe non è più avviata
				setStarted(false);
			}
		}
	}
	
	/**
	 * Accoda un nuovo messaggio alla Pipe.
	 * Se la Pipe risulta piena, la chiamata e' bloccante
	 */
	public void enqueue(E parMessageToEnqueue) {
		synchronized (pMessageQueue) {
			
			// Verifica che la pipe sia avviata
			if(!this.isStarted())
				return;
			
			// Se e' prevista una lunghezza massima della coda ..
			if(pMessageQueueMaxLenght > 0) {
				
				// .. e questa lunghezza e' stata raggiunta ..
				while(pMessageQueue.size() >= pMessageQueueMaxLenght) {
					
					// .. si mette in attesa
					try {
						pMessageQueue.wait();
					}
					catch(Exception parException) { 
						// TODO: mettere il relay di InterruptedException
					}
				}
			}
				
			// Aggiunge il nuovo messaggio alla coda
			pMessageQueue.add(parMessageToEnqueue);
			
			// Segnala agli Waiter in attesa che ci sono messaggi in coda
			pWaitable.notifyWaiter();				
		}
	}
	
	/**
	 * Estrae dalla pipe il messaggio meno recente.
	 * Se la Pipe e' vuota, la chiamata ritorna null.
	 * @return oggetto recuperato dalla Pipe, null se la Pipe e' vuota.
	 */
	public E dequeue() {
		E tDequeuedMessage = null;
		
		synchronized(pMessageQueue) {
			
			// Verifica che la pipe sia avviata
			if(!isStarted())
				return null;
			
			// Se la pipe non e' vuota..
			if(!pMessageQueue.isEmpty()) {
				// .. toglie dalla pipe il messaggio meno recente
				tDequeuedMessage = pMessageQueue.poll();
			
				// .. risveglia eventuali thread in attesa di poter inserire
				pMessageQueue.notifyAll();
				
				// .. se la pipe si e' svuotata, resetta il segnale nell'oggetto Waitable
				if(pMessageQueue.isEmpty()) {
					pWaitable.resetWaiter();
				}
			}			
		}
		
		// Restituisce il messaggio che e' stato tolto dalla pipe
		return tDequeuedMessage;
	}
	
	/**
	 * Restituisce un riferimento all'oggetto Waitable
	 */
	public Waitable<Pipe<E>> getWaitable() {	
		return pWaitable;
	}
}
